package com.rpc;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * Message properties:AMQP 0-9-1协议预先定义了14个消息属性，大多数的消息属性不怎么用，
 * 常用的消息属性有如下一些
 * deliverMode：值设置为2表示持久消息，其它值表示临时消息
 * contentType：描述编码类型
 * replyTo：一个回调队列的名字
 * correlationId：请求相关的rpc响应Id
 */
public class RPCClient implements AutoCloseable {

    private Connection connection;
    private Channel channel;
    private String requestQueueName = RPCServer.RPC_QUEUE_NAME;

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();
        //声明了一个临时队列，专门存放计算结果的
        String replyQueueName = channel.queueDeclare().getQueue();
        //一个客户端只有一个回调队列，通过correlationId来区分不同的请求
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        // 发布计算请求，消息中带上了correlationId与响应队列名的信息
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        //获取计算结果，要判断是否与发送时产生的correlationId是一样的
        //由于是异步消费的，所以暂时把结果放置到阻塞队列，有数据后导致后面的take方法调用解除阻塞
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            //在后台rabbitmq的消费线程中获取服务端的响应
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), "UTF-8"));
            }
        }, consumerTag -> {});

        //System.out.println("在主线程中运行：从阻塞队列中提取结果");
        //take如果取不到数据会阻塞
        String result = response.take();
        //取消channel，会调用 Consumer.handleCancelOk，相当于删除消费者
        channel.basicCancel(ctag);

        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
}